#include "TranCodec.h"

using namespace x2lib;

#pragma region TranCodec

TranCodec::TranCodec(bool isClient, Listener* pListener, uint8_t mgCode[16], int nPackSize, bool syncExec):
    m_isClient(isClient),
    m_pListener(pListener),
    m_syncExec(syncExec)
{
    m_isRunning = false;
    m_pMyChannel = nullptr;
    memcpy(m_mgCode, mgCode, sizeof(m_mgCode));
    m_pSendThread = nullptr;
    m_pRecvThread = nullptr;
    m_pAcceptThread = nullptr;
}

TranCodec::~TranCodec()
{

}

bool TranCodec::Init(void* userPtr)
{
    if (m_isClient)
    {
        m_pMyChannel = connectServer(userPtr);
        if (m_pMyChannel)
        {
            m_pListener->OnConnChanged(m_pMyChannel);
        }
    }
    else
    {
        m_pMyChannel = prepareServer(userPtr);
    }
    if (m_pMyChannel != nullptr)
    {
        m_vecpChannel.push_back(m_pMyChannel);
    }
    return (m_pMyChannel != nullptr);
}

bool TranCodec::Execute(int reConnWaitsec, int reConnTimes)
{
    if (m_isRunning || !m_pMyChannel) { return false; }

    m_isRunning = true;
    m_pSendThread = new std::thread([this]()-> void {
        do
        {
            bool isBusy = false;
            for (auto& it : m_vecpChannel)
            {
                if (!m_isClient)
                {
                    if (it == m_pMyChannel) { continue; }
                }

                if (it->conn != true) { continue; }

                int status = trySend(it, false);
                if (0 == status)
                {
                    it->conn = false;
                    m_pListener->OnConnChanged(it);
                    closeChannel(it);
                }
                else if (1 == status)
                {
                }
                else
                {
                    isBusy |= true;
                }
            }
            if (!isBusy)
            {
                std::this_thread::sleep_for(std::chrono::milliseconds(10));
            }
        } while (m_isRunning);
        });

    m_pRecvThread = new std::thread([this]()-> void {
        do
        {
            bool isBusy = false;
            for (auto& it : m_vecpChannel)
            {
                if (!m_isClient)
                {
                    if (it == m_pMyChannel) { continue; }
                }

                if (it->conn != true) { continue; }

                int status = tryRecv(it, false);
                if (0 == status)
                {
                    it->conn = false;
                    m_pListener->OnConnChanged(it);
                    closeChannel(it);
                }
                else if (1 == status)
                {
                }
                else
                {
                    isBusy |= true;
                }
            }
            if (!isBusy)
            {
                std::this_thread::sleep_for(std::chrono::milliseconds(10));
            }
        } while (m_isRunning);
        });

    if (!m_isClient)
    {
        m_pAcceptThread = new std::thread([this]()-> void {
            do
            {
                AbsChannel* pChannel = acceptClients();
                if (pChannel)
                {
                    pChannel->conn = true;
                    m_pListener->OnConnChanged(pChannel);
                    m_mtxChannel.Lock();
                    m_vecpChannel.push_back(pChannel);
                    m_mtxChannel.Unlock();
                }
            } while (m_isRunning);
            });
    }

    if (m_syncExec)
    {
        m_pSendThread->join();
        m_pRecvThread->join();
        delete m_pSendThread;
        delete m_pRecvThread;
        if (!m_isClient)
        {
            m_pAcceptThread->join();
            delete m_pAcceptThread;
        }
    }
    return 0;
}

void TranCodec::UnInit(void* userPtr)
{
    m_isRunning = false;
    if (!m_syncExec)
    {
        m_pSendThread->join();
        m_pRecvThread->join();
        delete m_pSendThread;
        delete m_pRecvThread;
        if (!m_isClient)
        {
            m_pAcceptThread->join();
            delete m_pAcceptThread;
        }
    }

    for (auto& it : m_vecpChannel)
    {
        delete it;
    }
    m_vecpChannel.clear();
    this->releaseServer(userPtr);
}

TranCodec::AbsChannel* TranCodec::GetMyChannel()
{
    return m_pMyChannel;
}

TranCodec::AbsChannel* TranCodec::FindChannel(std::function<bool(const AbsChannel*)> condition)
{
    AbsChannel* pChannel = nullptr;

    m_mtxChannel.Lock();
    for (auto it : m_vecpChannel)
    {
        if (condition(it)) 
        {
            pChannel = it;
            break;
        }
    }
    m_mtxChannel.Unlock();

    return pChannel;
}

uint32_t TranCodec::PostData(AbsChannel* pChannel, uint32_t nMsgId, bool action, uint8_t retCode, const void* pData, uint32_t nData, bool isWait) 
{
    if (pChannel == nullptr) { return 0; }
    bool isSucc = true;
    TranPack* pTranPack = (TranPack*)malloc(pChannel->uPackSize);
    pTranPack->incDbgId();
    for (uint32_t index = 0; index <= nData; index += pChannel->uPackSize)
    {
        pTranPack->reset(m_mgCode);
        uint32_t nBytes = MIN(MIN(pChannel->uPackSize, nData - index), nData);
        pTranPack->setPack(nMsgId, action, 0, index, nData, (uint8_t*)pData + index, (uint16_t)nBytes);

        if (!pushData((uint8_t*)pTranPack, TranPack::MIN_SIZE + pTranPack->Head.nBytes, pChannel, true, isWait)) 
        {
            free(pTranPack);
            return 0;
        }
    }

    uint32_t uDbgId = pTranPack->Head.uDbgId;
    free(pTranPack);
    return uDbgId;
}

uint32_t TranCodec::SendData(AbsChannel* pChannel, uint32_t nMsgId, bool action, uint8_t retCode, const void* pData, uint32_t nData, void* pDataout, uint32_t nDataout, bool isWait)
{
    TranPackHead tranPackHead = { 0 };
    uint32_t uDbgId = PostData(pChannel, nMsgId, action, retCode, pData, nData, isWait);
    if (uDbgId == 0)
    {
        return 0;
    }
    m_setSyncId.insert(uDbgId);
    while(pChannel->conn)
    {
        m_signSend.Acquire();

        if (!pChannel->pPackRxed->isValid() || nDataout < pChannel->pPackRxed->Head.nTotal)
        {
            m_setSyncId.erase(pChannel->pPackRxed->Head.uDbgId);
            return 0;
        }

        memcpy(&((uint8_t*)pDataout)[pChannel->pPackRxed->Head.iSeek], pChannel->pPackRxed->bodyAddr(), pChannel->pPackRxed->bodySize());
        if (pChannel->pPackRxed->Head.iSeek + pChannel->pPackRxed->Head.nBytes >= pChannel->pPackRxed->Head.nTotal)
        {
            m_setSyncId.erase(pChannel->pPackRxed->Head.uDbgId);
            break;
        }
    }

    return pChannel->pPackRxed->Head.nTotal;
}

bool TranCodec::pushData(uint8_t* pData, uint32_t nData, AbsChannel* pChannel, bool isTx, bool isWait) 
{
    AbsChannel::Buffer* buffer = isTx ? pChannel->pPackTxBuffer : pChannel->pPackRxBuffer;
    
    while (true) 
    {
        if (nData <= buffer->spaceBytes()) 
        {
            break;
        }
        
        if (!isWait) { return false; }
        
        std::this_thread::sleep_for(std::chrono::milliseconds(100));
    }

    for (uint32_t i = 0; i < nData; ++i) 
    {
        (*buffer)[buffer->endIndex] = pData[i];
        buffer->endIndex = (uint16_t)(buffer->endIndex + 1) % buffer->totalBytes();
    }

    return true;
}

TranCodec::TranPack* TranCodec::pullData(AbsChannel* pChannel, bool isTx, bool isWait) 
{
    AbsChannel::Buffer* buffer = isTx ? pChannel->pPackTxBuffer : pChannel->pPackRxBuffer;
    TranPack* tranPack = isTx ? pChannel->pPackTxed : pChannel->pPackRxed;

    while (true) 
    {
        if (buffer->filledBytes() < TranPack::MIN_SIZE) 
        {
            if (!isWait) { break; }
            std::this_thread::sleep_for(std::chrono::milliseconds(100));
            continue;
        }

        // 查找魔术字位置
        int hashIdx = -1;
        for (uint32_t i = buffer->beginIndex; i != buffer->endIndex && hashIdx < 0; ++i) 
        {
            hashIdx = i;
			for (size_t j = 0; j < sizeof(m_mgCode); ++j)
			{
				if ((*buffer)[(i + j) % buffer->totalBytes()] != m_mgCode[j])
				{
					hashIdx = -1;
					break;
				}
			}
        }

        // 丢弃1字节
        if (hashIdx < 0) 
        {
            if (!isWait) { break; }
            buffer->beginIndex += 1;
            continue;
        }

        uint8_t* pData = (uint8_t*)tranPack;
        // 使头有效
        uint32_t bodyIdx = hashIdx;
        for (uint32_t j = 0; j < TranPack::MIN_SIZE; ++j) 
        {
            pData[j] = (*buffer)[bodyIdx];
            bodyIdx = (bodyIdx + 1) % buffer->totalBytes();
        }

        if (buffer->filledBytes() < TranPack::MIN_SIZE + tranPack->Head.nBytes)
        {
            if (!isWait) { break; }
            std::this_thread::sleep_for(std::chrono::milliseconds(100));
            continue;
        }
        buffer->beginIndex = bodyIdx;
        for (uint32_t j = 0; j < tranPack->Head.nBytes; ++j) 
        {
            pData[TranPack::MIN_SIZE+j] = (*buffer)[buffer->beginIndex];
            buffer->beginIndex = (buffer->beginIndex + 1) % buffer->totalBytes();
        }

        return tranPack;
    }

    return nullptr;
}

#if 0 // 示例（Socket）
int TranCodec::ApiSend(AbsChannel* pChannel, void* pData, int nData, int* pnBytes)
{
    int iRet = send(fd, (char*)pData, nData, 0);
    *pnByte = iRet > 0 ? iRet : 0;
#ifdef __X2LIB_WIN32__
    if (iRet < 1 && WSAEWOULDBLOCK != WSAGetLastError())
#elif defined __X2LIB_LINUX__
    if ((iRet == -1 && (errno == EBADF || errno == ECONNRESET)) // 主动关闭
        || (iRet == 0 && errno == EWOULDBLOCK) // 被动关闭
        || (iRet < 0 && (errno != EWOULDBLOCK) && (errno != EINPROGRESS))) // 正在接收
#endif
    {
        return false;
    }

    return true;
}

bool TranCodec::ApiRecv(void* pData, int nData, int* pnByte)
{
    // iRet=0 indicate socket closed,iRet<0 indicate send error
    int iRet = recv(fd, (char*)pData, nData, 0);
    *pnByte = iRet > 0 ? iRet : 0;
#ifdef __X2LIB_WIN32__
    if (iRet < 1 && WSAEMSGSIZE != WSAGetLastError())
#elif defined __X2LIB_LINUX__
    if ((iRet == -1 && (errno == EBADF || errno == ECONNRESET)) // 主动关闭
        || (iRet == 0 /*&& errno == EWOULDBLOCK*/) // 被动关闭
        || (iRet < 0 && (errno != EWOULDBLOCK) && (errno != EINPROGRESS))) // 正在接收
#endif
    {
        return false;
    }
    return true;
}
#endif

int TranCodec::trySend(AbsChannel* pChannel, bool isWait)
{
    TranPack* pTranPack = pullData(pChannel, true, isWait);
    if (!pTranPack) { return 1; }

    uint8_t* pData = (uint8_t*)pTranPack->memAddr();
    uint16_t nData = pTranPack->memSize();

    m_pListener->OnBytesEnSend(pData, nData);

    for (uint32_t i = 0; i < nData;)
    {
        uint32_t nSent = 0;
        if (!ApiSend(pChannel, &pData[i], nData - i, &nSent))
        {
            return 0;
        }
        i += nSent;
    }

    m_pListener->OnBytesDeRecv(pData, nData);
    m_pListener->OnPackSended(pChannel);

    return 2;
}

int TranCodec::tryRecv(AbsChannel* pChannel, bool isWait)
{
    int iRet = 0;
    uint32_t nData = pChannel->pPackRxBuffer->spaceBytes();
    uint8_t* pData = (uint8_t*)malloc(nData);
    do
    {
        uint32_t nRecv = 0;
        if (!ApiRecv(pChannel, pData, nData, &nRecv))
        {
            iRet = 0;
            break;
        }

        m_pListener->OnBytesDeRecv(pData, nRecv);

        pushData(pData, nRecv, pChannel, false, isWait);

        if (!pullData(pChannel, false, isWait))
        {
            iRet = 1;
            break;
        }

        void* ppRepData = nullptr;
        uint32_t nRepData = 0;
        uint8_t retCode = m_pListener->OnPackRecved(pChannel, &ppRepData, &nRepData);

        if (m_setSyncId.find(pChannel->pPackRxed->Head.uDbgId) != m_setSyncId.end())
        { // 通知SendData已收到数据
            m_signSend.Release();
        }

        if (!pChannel->pPackRxed->Head.action)
        { // response
            PostData(pChannel, pChannel->pPackRxed->Head.nMsgId, true, retCode, ppRepData, nRepData, true);
        }

        if (ppRepData) { free(ppRepData); }

        iRet = 2;
    } while (0);
    free(pData);

    return iRet;
}

#pragma endregion

